package com.example.springbootkafka.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;

import java.util.HashMap;
import java.util.Map;

/**
 * @author Bruce Lee
 * @Classname KafkaConfig
 * @Description 生产者/消费组相关配置类
 * @Date 2023-11-22
 */
@Slf4j
@EnableKafka
@SpringBootConfiguration
public class KafkaConfig {


    /**
     * producer configuration
     *
     * @return
     */
    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        // 发送重试次数，默认2147483647，接近无限大
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        // 一个批次发送的大小，默认16KB，超过这个大小就会发送数据
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 一个批次最长等待多久就发送数据，默认0，即马上发送
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5000);
        // 生产者内存缓冲区大小; 默认33554432bytes=32MB
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }


    /**
     * consumer configuration
     *
     * @return
     */
    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        // 消费者组ID
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "0");
        // 是否自动提交偏移，默认true
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // 自动提交间隔，默认5秒；从开始消费一条数据到业务结束，必须在5秒内完成，否则会造成提前提交偏移量，如果出现事务失败，将会漏掉该条消费
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
        // 会话超时时间；若超时未收到相应心跳，则会将该消费者从组内移除并启动重新平衡
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    ConcurrentKafkaListenerContainerFactory<String, String>
    kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        // 传入消费者工厂
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }


    /**
     * kafka template configuration
     *
     * @return
     */
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        // 传入生产者工厂
        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
        return kafkaTemplate;
    }

}
